c60aebe0311ac1bb46b34a34e8d94ec66a79db60,clients/unshaded/src/main/java/tachyon/client/UfsUtils.java,UfsUtils,loadUfs,#TachyonFS#TachyonURI#TachyonURI#PrefixList#TachyonConf#,97

Before Change


      if (ufs.isFile(ufsPath.toString())) { // TODO(hy): Fix path matching issue.
        TachyonURI tfsPath = buildTFSPath(directoryName, ufsAddrRootPath, ufsPath);
        LOG.debug("Loading ufs. tfs path = {}.", tfsPath);
        if (tfs.exist(tfsPath)) {
          LOG.debug("File {} already exists in Tachyon.", tfsPath);
          continue;
        }

After Change


   * @throws IOException when an event that prevents the operation from completing is encountered
   */
  @Deprecated
  public static void loadUfs(TachyonFileSystem tfs, TachyonURI tachyonPath, TachyonURI
      ufsAddrRootPath, PrefixList excludePathPrefix, TachyonConf tachyonConf) throws IOException,
      TachyonException {
    LOG.info("Loading to {} {} {}", tachyonPath, ufsAddrRootPath, excludePathPrefix);
    try {
      // resolve and replace hostname embedded in the given ufsAddress/tachyonAddress
      ufsAddrRootPath = NetworkAddressUtils.replaceHostName(ufsAddrRootPath);
      tachyonPath = NetworkAddressUtils.replaceHostName(tachyonPath);
    } catch (UnknownHostException e) {
      LOG.error("Failed to resolve hostname", e);
      throw new IOException(e);
    }

    Pair<String, String> ufsPair = UnderFileSystem.parse(ufsAddrRootPath, tachyonConf);
    String ufsAddress = ufsPair.getFirst();
    String ufsRootPath = ufsPair.getSecond();

    LOG.debug("Loading ufs, address:{}; root path: {}", ufsAddress, ufsRootPath);

    // create the under FS handler (e.g. hdfs, local FS, s3 etc.)
    UnderFileSystem ufs = UnderFileSystem.get(ufsAddress, tachyonConf);

    if (!ufs.exists(ufsAddrRootPath.toString())) {
      throw new FileNotFoundException("ufs path " + ufsAddrRootPath + " not found.");
    }

    // directory name to load, either the path parent or the actual path if it is a directory
    TachyonURI directoryName;
    if (ufs.isFile(ufsAddrRootPath.toString())) {
      if ((ufsRootPath == null) || ufsRootPath.isEmpty() || ufsRootPath.equals("/")) {
        directoryName = TachyonURI.EMPTY_URI;
      } else {
        int lastSlashPos = ufsRootPath.lastIndexOf('/');
        if (lastSlashPos > 0) {
          directoryName = new TachyonURI(ufsRootPath.substring(0, lastSlashPos)); // trim the slash
        } else {
          directoryName = TachyonURI.EMPTY_URI;
        }
      }
    } else {
      directoryName = tachyonPath;
    }

    if (!directoryName.equals(TachyonURI.EMPTY_URI)) {
      if (tfs.openIfExists(directoryName) == null) {
        LOG.debug("Loading ufs. Make dir if needed for '{}'.", directoryName);
        tfs.mkdir(directoryName);
      }
      // TODO(hy): Add the following.
      // if (tfs.mkdir(tfsRootPath)) {
      // LOG.info("directory " + tfsRootPath + " does not exist in Tachyon: created");
      // } else {
      // throw new IOException("Failed to create folder in Tachyon: " + tfsRootPath);
      // }
    }

    Queue<TachyonURI> ufsPathQueue = new LinkedList<TachyonURI>();
    if (excludePathPrefix.outList(ufsRootPath)) {
      ufsPathQueue.add(ufsAddrRootPath);
    }

    while (!ufsPathQueue.isEmpty()) {
      TachyonURI ufsPath = ufsPathQueue.poll(); // this is the absolute path
      LOG.debug("Loading: {}", ufsPath);
      if (ufs.isFile(ufsPath.toString())) { // TODO(hy): Fix path matching issue.
        TachyonURI tfsPath = buildTFSPath(directoryName, ufsAddrRootPath, ufsPath);
        LOG.debug("Loading ufs. tfs path = {}.", tfsPath);
        if (tfs.openIfExists(tfsPath) != null) {
          LOG.debug("File {} already exists in Tachyon.", tfsPath);
          continue;
        }
        long fileId = tfs.loadMetadata(tfsPath).getFileId();
        LOG.debug("Create tachyon file {} with file id {} and checkpoint location {}", tfsPath,
            fileId, ufsPath);
      } else { // ufsPath is a directory
        LOG.debug("Loading ufs. ufs path is a directory.");
        String[] files = ufs.list(ufsPath.toString()); // ufs.list() returns relative path
        if (files != null) {
          for (String filePath : files) {
            if (filePath.isEmpty()) { // Prevent infinite loops
              continue;
            }
            LOG.debug("Get: {}", filePath);
            String aPath = PathUtils.concatPath(ufsPath, filePath);
            String checkPath = aPath.substring(ufsAddrRootPath.toString().length());
            if (checkPath.startsWith(TachyonURI.SEPARATOR)) {
              checkPath = checkPath.substring(TachyonURI.SEPARATOR.length());
            }
            if (excludePathPrefix.inList(checkPath)) {
              LOG.debug("excluded: {}", checkPath);
            } else {
              ufsPathQueue.add(new TachyonURI(aPath));
            }
          }
        }
        // ufsPath is a directory, so only concat the tfsRoot with the relative path
        TachyonURI tfsPath = new TachyonURI(PathUtils.concatPath(
            tachyonPath, ufsPath.getPath().substring(ufsAddrRootPath.getPath().length())));
        LOG.debug("Loading ufs. ufs path is a directory. tfsPath = {}.", tfsPath);
        if (tfs.openIfExists(tfsPath) == null) {
          LOG.debug("Loading ufs. ufs path is a directory. make dir = {}.", tfsPath);
          tfs.mkdir(tfsPath);
          // TODO(hy): Add the following.